package com.lkd.service.impl;

import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.JsonNode;
import com.google.common.base.Strings;
import com.google.common.collect.Lists;
import com.lkd.common.VMSystem;
import com.lkd.config.ConsulConfig;
import com.lkd.config.TopicConfig;
import com.lkd.contract.OrderCheck;
import com.lkd.contract.VendoutContract;
import com.lkd.contract.VendoutData;
import com.lkd.dao.OrderDao;
import com.lkd.emq.MqttProducer;
import com.lkd.entity.OrderEntity;
import com.lkd.exception.LogicException;
import com.lkd.feign.UserService;
import com.lkd.feign.VMService;
import com.lkd.service.OrderService;
import com.lkd.utils.DistributedLock;
import com.lkd.utils.JsonUtil;
import com.lkd.vo.*;
import lombok.extern.slf4j.Slf4j;
import org.elasticsearch.action.search.SearchRequest;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.index.query.BoolQueryBuilder;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.index.query.RangeQueryBuilder;
import org.elasticsearch.search.SearchHit;
import org.elasticsearch.search.SearchHits;
import org.elasticsearch.search.aggregations.AggregationBuilder;
import org.elasticsearch.search.aggregations.AggregationBuilders;
import org.elasticsearch.search.aggregations.Aggregations;
import org.elasticsearch.search.aggregations.BucketOrder;
import org.elasticsearch.search.aggregations.bucket.terms.Terms;
import org.elasticsearch.search.builder.SearchSourceBuilder;
import org.elasticsearch.search.sort.SortOrder;
import org.redisson.api.RLock;
import org.redisson.api.RedissonClient;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Service;

import java.io.IOException;
import java.math.BigDecimal;
import java.math.RoundingMode;
import java.time.Duration;
import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.List;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

@Service
@Slf4j
public class OrderServiceImpl extends ServiceImpl<OrderDao, OrderEntity> implements OrderService {


    @Override
    public OrderEntity getByOrderNo(String orderNo) {
        QueryWrapper<OrderEntity> qw = new QueryWrapper<>();
        qw.lambda()
                .eq(OrderEntity::getOrderNo,orderNo);
        return this.getOne(qw);
    }

    @Autowired
    private VMService vmService;

    @Autowired
    private UserService userService;

    @Autowired
    private RedissonClient redissonClient;


    @Override
    public OrderEntity createOrder(PayVO payVO) {
        OrderEntity orderEntity = new OrderEntity();
        //分布式锁
        //锁设备   redisson

        try {
            if (!vmService.hasCapacity(payVO.getInnerCode()
                    , Long.valueOf(payVO.getSkuId()))) {
                throw new LogicException("该商品已售空");
            }

            orderEntity.setOpenId(payVO.getOpenId());//openId
            orderEntity.setPayStatus(VMSystem.PAY_STATUS_NOPAY);//支付状态
            orderEntity.setOrderNo(payVO.getInnerCode() + System.nanoTime());//订单编号

            //根据售货机编号查询售货机信息 ,并复制给订单对象
            VmVO vm = vmService.getVMInfo(payVO.getInnerCode());
            BeanUtils.copyProperties(vm, orderEntity);
            orderEntity.setAddr(vm.getNodeAddr());

            //根据商品编号查询商品信息，并复制给订单对象
            SkuVO skuVo = vmService.getSku(payVO.getSkuId());
            BeanUtils.copyProperties(skuVo, orderEntity);
            orderEntity.setAmount(skuVo.getRealPrice());//价格
            orderEntity.setStatus(VMSystem.ORDER_STATUS_CREATE);

            //设置合作商账单分账金额
            PartnerVO partner = userService.getPartner(vm.getOwnerId());
            BigDecimal bg = new BigDecimal(skuVo.getRealPrice());
            int bill = bg.multiply(new BigDecimal(partner.getRatio())).divide(new BigDecimal(100), 0, RoundingMode.HALF_UP).intValue();
            orderEntity.setBill(bill);

            this.save(orderEntity);
            //发送延迟消息
            sendDelayedMessage(orderEntity.getOrderNo());
        }catch (Exception e) {
            log.error("系统异常", e);
        }
        return orderEntity;
    }


    /**
     *
     * 发送延迟消息
     * @param orderNo
     */
    private void sendDelayedMessage(String orderNo) {
        try {
            //将订单放到延迟队列中，10分钟后检查支付状态！！！！！！！！！！！！！！！！！！
            OrderCheck orderCheck = new OrderCheck();
            orderCheck.setOrderNo(orderNo);
            mqttProducer.send("$delayed/600/" + TopicConfig.ORDER_CHECK_TOPIC, 2, orderCheck);
        }catch (Exception e) {
            log.error("发送延迟消息异常",e);
        }
    }

    @Autowired
    private MqttProducer mqttProducer;

    @Override
    public boolean vendout(String orderNo) {
        //根据订单号查询订单
        OrderEntity orderEntity = getByOrderNo(orderNo);
        //修改订单状态为 “支付完成”
        orderEntity.setStatus(VMSystem.ORDER_STATUS_PAYED);
        orderEntity.setPayStatus(VMSystem.PAY_STATUS_PAYED);
        this.updateById(orderEntity);

        //封装协议中的数据
        var reqData = new VendoutData();
        reqData.setOrderNo(orderEntity.getOrderNo());
        reqData.setSkuId(orderEntity.getSkuId());
        //封装协议
        var contract = new VendoutContract();
        contract.setVendoutData(reqData);
        contract.setInnerCode(orderEntity.getInnerCode());
        //向售货机微服务  发送出货请求
        try {
            mqttProducer.send( TopicConfig.VMS_VENDOUT_TOPIC,2,contract);
        } catch (JsonProcessingException e) {
            log.error("send vendout req error.",e);
            return false;
        }
        return true;
    }


    @Autowired
    private RestHighLevelClient esClient;

    @Override
    public Pager<OrderVO> search(Integer pageIndex, Integer pageSize, String orderNo, String openId, String startDate, String endDate) {
        SearchRequest searchRequest = new SearchRequest("order");
        SearchSourceBuilder sourceBuilder = new SearchSourceBuilder();
        BoolQueryBuilder boolQueryBuilder = QueryBuilders.boolQuery();
        //根据订单号查询
        if(!Strings.isNullOrEmpty(orderNo)) {
            boolQueryBuilder.must(QueryBuilders.termQuery("order_no", orderNo));
        }
        //根据openId查询
        if(!Strings.isNullOrEmpty(openId)){
            boolQueryBuilder.must(QueryBuilders.termQuery("open_id",openId));
        }
        //根据时间范围查询
        if(!Strings.isNullOrEmpty(startDate) && !Strings.isNullOrEmpty(endDate)){
            RangeQueryBuilder rangeQueryBuilder = QueryBuilders.rangeQuery("update_time");
            rangeQueryBuilder.gte(startDate);
            rangeQueryBuilder.lte(endDate);
            boolQueryBuilder.must(rangeQueryBuilder);
        }
        //按照最后更新时间由近到远的排序规则排序
        sourceBuilder.from((pageIndex -1)*pageSize);
        sourceBuilder.size(pageSize);
        sourceBuilder.sort("update_time", SortOrder.DESC);
        sourceBuilder.trackTotalHits(true);
        sourceBuilder.query(boolQueryBuilder);
        searchRequest.source(sourceBuilder);
        try {
            SearchResponse searchResponse = esClient.search(searchRequest, RequestOptions.DEFAULT);
            SearchHits hits = searchResponse.getHits();
            SearchHit[] searchHits = hits.getHits();
            List<OrderVO> orderList = Lists.newArrayList();
            for(SearchHit hit : searchHits){
                String hitResult = hit.getSourceAsString();
                var order = new OrderVO();
                JsonNode jsonNode = JsonUtil.getTreeNode(hitResult);
                order.setId(jsonNode.findPath("id").asLong());
                order.setStatus(jsonNode.findPath("status").asInt());
                order.setBill(jsonNode.findPath("bill").asInt());
                order.setOwnerId(jsonNode.findPath("owner_id").asInt());
                order.setPayType(jsonNode.findPath("pay_type").asText());
                order.setOrderNo(jsonNode.findPath("order_no").asText());
                order.setInnerCode(jsonNode.findPath("inner_code").asText());
                order.setSkuName(jsonNode.findPath("sku_name").asText());
                order.setSkuId(jsonNode.findPath("sku_id").asLong());
                order.setPayStatus(jsonNode.findPath("pay_status").asInt());
                order.setBusinessName(jsonNode.findPath("business_name").asText());
                order.setBusinessId(jsonNode.findPath("business_id").asInt());
                order.setRegionId(jsonNode.findPath("region_id").asLong());
                order.setRegionName(jsonNode.findPath("region_name").asText());
                order.setPrice(jsonNode.findPath("price").asInt());
                order.setAmount(jsonNode.findPath("amount").asInt());
                order.setAddr(jsonNode.findPath("addr").asText());
                order.setOpenId(jsonNode.findPath("open_id").asText());
                order.setCreateTime(LocalDateTime.parse(jsonNode.findPath("create_time").asText(),DateTimeFormatter.ISO_DATE_TIME));
                order.setUpdateTime(LocalDateTime.parse(jsonNode.findPath("update_time").asText(),DateTimeFormatter.ISO_DATE_TIME));
                orderList.add(order);
            }
            Pager<OrderVO> pager = new Pager<>();
            pager.setTotalCount(searchResponse.getHits().getTotalHits().value);

            pager.setPageSize(searchResponse.getHits().getHits().length);
            pager.setCurrentPageRecords(orderList);
            pager.setPageIndex(pageIndex);

            return pager;
        } catch (IOException e) {
            log.error("查询es失败",e);
            return Pager.buildEmpty();
        }
    }


    @Override
    public List<Long> getTop10Sku(Integer businessId) {

        SearchRequest searchRequest=new SearchRequest("order");
        SearchSourceBuilder sourceBuilder=new SearchSourceBuilder();
        //查询条件：最近三个月

        RangeQueryBuilder rangeQueryBuilder = QueryBuilders.rangeQuery("update_time");
        rangeQueryBuilder.gte( LocalDateTime.now().plusMonths(-3).format(  DateTimeFormatter.ISO_DATE_TIME )  );
        rangeQueryBuilder.lte( LocalDateTime.now().format(  DateTimeFormatter.ISO_DATE_TIME )  );

        BoolQueryBuilder boolQueryBuilder=QueryBuilders.boolQuery();
        boolQueryBuilder.must( rangeQueryBuilder );

        boolQueryBuilder.must( QueryBuilders.termQuery("business_id",businessId) );
        sourceBuilder.query(boolQueryBuilder);

        AggregationBuilder orderAgg = AggregationBuilders.terms("sku").field("sku_id")
                .subAggregation(AggregationBuilders.count("count").field("sku_id"))
                .order(BucketOrder.aggregation("count", false))
                .size(10);

        sourceBuilder.aggregation(orderAgg);
        searchRequest.source(sourceBuilder);

        try {
            SearchResponse searchResponse = esClient.search(searchRequest, RequestOptions.DEFAULT);
            Aggregations aggregations = searchResponse.getAggregations();
            if(aggregations==null ) return  Lists.newArrayList();

            var term = (Terms)aggregations.get("sku");
            var buckets = term.getBuckets();

            return buckets.stream().map(  b->   Long.valueOf( b.getKey().toString() ) ).collect(Collectors.toList());

        } catch (IOException e) {
            e.printStackTrace();
            return Lists.newArrayList();
        }

    }


}
